/******************************************************************************/
/* @ All Rights Reserved Copyright (C) 2014 broada,Ltd.                       */
/*                                                                            */
/*  NAME      = kafka_comm.h                                                  */
/*  NOTE      =                                                               */
/*  DATE      = 2015/10/18 by zhengxd                                         */
/******************************************************************************/
#ifndef _KAFKA_COMM_H_
#define _KAFKA_COMM_H_

#define  KAFKA_CHAR      '1'
#define  KAFKA_SHORT     '2'
#define  KAFKA_INT       '3'
#define  KAFKA_LONG      '4'
#define  KAFKA_STRINT    '5'
#define  KAFKA_BYTES     '6'

#define KAFKA_REPORT_KEY_SESSION_TOPIC     APM_REPORT_KEY_SESSION ".mq.topic"
#define KAFKA_REPORT_KEY_REQ_TOPIC         APM_REPORT_KEY_REQ".mq.topic"
#define KAFKA_REPORT_KEY_REQ_METHOD        APM_REPORT_KEY_REQ".mq.method"
#define KAFKA_REPORT_KEY_REQ_TYPE          APM_REPORT_KEY_REQ".mq.message.type"
#define KAFKA_REPORT_KEY_REQ_TEXT          APM_REPORT_KEY_REQ".mq.message.text"
#define KAFKA_REPORT_KEY_RESP_SIZE         APM_REPORT_KEY_RESP"mq.message.size"
#define KAFKA_REPORT_KEY_RESP_TYPE         APM_REPORT_KEY_RESP".mq.message.type."
#define KAFKA_REPORT_KEY_RESP_TEXT         APM_REPORT_KEY_RESP".mq.message.text."

/* message set信息 */
typedef struct __st_kafka_message_set_info {

    char *p_value;       /* bytes */
} Kafka_message_set_info;

/* partition信息 */
typedef struct __st_kafka_partition_info {

    Frlist_header st_messages;      //Kafka_message_set_info
} Kafka_partition_info;

/* topic信息 */
typedef struct __st_kafka_topic_info {

    char *p_topic_name;
    Frlist_header st_partitions;     //Kafka_partition_info
} Kafka_topic_info;

void kafka_dbg(void);

void kafka_free(char **pp_buffer);

int kafka_get_short(Frbyte_Packet *p_st_data, int *p_i_len);

int kafka_get_int(Frbyte_Packet *p_st_data, int *p_i_len);

int kafka_get_string(Frbyte_Packet *p_st_data, char **pp_out_str);

int kafka_get_bytes(Frbyte_Packet *p_st_data, int i_attributes
                    , char **pp_out_bytes, unsigned int *p_ui_size);

int kafka_skip_string(Frbyte_Packet *p_st_data);

int kafka_skip_bytes(Frbyte_Packet *p_st_data);

int kafka_parse_message_set(Frbyte_Packet *p_st_data
                           , unsigned int ui_message_set_size
                           , Frlist_header *p_st_message_set);

void kafka_message_set_node_free(Frlist_node *p_st_node);

void kafka_message_set_free(Kafka_message_set_info **pp_st_message_set);

void kafka_partition_node_free(Frlist_node *p_st_node);

void kafka_partition_free(Kafka_partition_info **pp_st_partition);

void kafka_topic_free(Kafka_topic_info **pp_st_topic);

void kafka_topic_node_free(Frlist_node *p_st_node);

void kafka_destroy_request(L7_req_kafka **pp_st_req);

void kafka_destroy_response(L7_resp_kafka **pp_st_resp);

void kafka_init_fragment(L7_fragment_kafka *p_st_fragment);

void kafka_destroy_fragment(L7_fragment_kafka **pp_st_fragment);

unsigned long kafka_get_interval_time(struct timeval *p_st_start_time
                                     , struct timeval *p_st_end_time);

int kafka_get_msg_version(Frbyte_Packet *p_st_data
                          , unsigned int ui_size
                          , int *p_i_msg_ver);

int kafka_parse_messages(Frbyte_Packet *p_st_data
                         , unsigned int ui_size
                         , int i_msg_ver
                         , Frlist_header *p_st_messages);

int kafka_skip_arr(Frbyte_Packet *p_st_data
                   , unsigned int ui_len
                   , char *p_value);

#endif

